/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.samples.periodic;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.StringTokenizer;

import cgl.granules.dataset.DatasetException;
import cgl.granules.exception.CommunicationsException;
import cgl.granules.exception.DeploymentException;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceException;
import cgl.granules.operation.ProgressTracker;
import cgl.granules.operation.InstanceDeployer;

/**
 * The deployer for the example that demonstrates using Granules to perform
 * operations at regular intervals on a distributed set of machines. .
 * 
 * @author Shrideep Pallickara
 * 
 */
public class PeriodicDeployer extends InstanceDeployer {
    private ProgressTracker progressTracker;


    /**
     * Initializes the deployer.
     * 
     * @param streamingProperties
     *            The properties for streaming
     * @throws CommunicationsException
     *             If there are network problems during communications
     * @throws IOException
     *             If there are IO problems
     * @throws MarshallingException
     *             If there are marshalling erros
     * @throws DeploymentException
     *             If there are deployment related problems
     */
    public PeriodicDeployer(Properties streamingProperties)
        throws CommunicationsException, IOException, MarshallingException,
        DeploymentException {
        initialize(streamingProperties);
    }


    /**
     * Retrieves the ID of the submitted Job
     * 
     * @return the jobId
     */
    public String getJobId() {
        if (progressTracker != null) {
            return progressTracker.getIdentifier();
        }
        return null;
    }


    /**
     * Initializes a <code>PerioidicMap</code> instance based on the specified
     * arguments.
     * 
     * @param reduce
     *            The reducer that would be associated with the map.
     * @param mapNum
     *            The map instance number
     * @return The initialized <code>PerioidicMap</code> instance
     * @throws MapReduceException
     *             If there are problems related to Map/Reduce roles within the
     *             application.
     */
    private PeriodicMap initializePeriodicMap(PeriodicReduce reduce, int mapNum)
        throws MapReduceException {
        PeriodicMap map = new PeriodicMap();
        map.setAsExecutePeriodically(4000);
        // map.setAsExecuteFixedNumberOfTimes(4);

        Properties props = new Properties();

        String value = mapNum + "";
        String property = "Worker Number";
        props.put(property, value);

        map.setProcessingDirectives(props);
        map.addReduce(reduce);

        return map;
    }


    /**
     * Prepares the set of Map/Reduce instances to be deployed on a set of
     * machines.
     * 
     * @param numOfMaps
     *            The number of Maps that need to be deployed.
     * 
     * @throws MapReduceException
     *             If there is a MapReduceExcecution because of problems when
     *             setting up the roles.
     * 
     * @throws CommunicationsException
     *             If there are network problems during communications
     * @throws DeploymentException
     *             If there are deployment related problems
     * @throws MarshallingException
     *             If there are marshalling problems.
     */
    public void prepareInstancesAndDeploy(int numOfMaps)
        throws MapReduceException, CommunicationsException,
        DeploymentException, MarshallingException {
        PeriodicReduce reduce = new PeriodicReduce();
        reduce.setAsExecuteWhenDataAvailable();

        PeriodicMap maps[] = new PeriodicMap[numOfMaps];

        for (int i = 0; i < numOfMaps; i++) {
            maps[i] = initializePeriodicMap(reduce, i);
        }

        resetTrackers();
        progressTracker = deployOperations(maps);
    }


    /**
     * @param args
     */
    public static void main(String[] args) {
        String module = "PeriodicDeployer.main()->";
        if (args.length < 2) {
            System.out.println("Usage: java cgl.granules.samples.periodic"
                + "PeriodicDeployer <hostname> <port> [numOfMaps (optional)]");
            System.exit(0);
        }
        String hostname = args[0];
        String portnum = args[1];
        int numOfMaps = 100;

        if (args.length > 2) {
            numOfMaps = Integer.parseInt(args[2]);
        }

        Properties streamingProperties = new Properties();
        streamingProperties.put("hostname", hostname);
        streamingProperties.put("portnum", portnum);

        String helpInfo =
            "\n\nThis is example demonstrates the periodicity functionality "
                + "using Granules. A set of operations will be deployed to be"
                + " executed periodically on a set of machines. "
                + "The current list of commands include: \n"
                + "submit [num_of_maps (optional)]  \n"
                + "status        : Lists the status of the all Jobs\n"
                + "refresh       : Refresh status of submitted job. Follow this"
                + " up with the status command to see status updates.\n"
                + "h             : List the set of commands \n\n";

        try {
            PeriodicDeployer deployer =
                new PeriodicDeployer(streamingProperties);

            StringTokenizer tok;
            InputStreamReader isr = new InputStreamReader(System.in);
            BufferedReader commandLine = new BufferedReader(isr);

            System.out.println(helpInfo);

            while (true) {
                String s = commandLine.readLine();
                if (s == null) {
                    System.out.println(module + "String is null!!!");
                    break;
                }

                if (s.startsWith("submit")) {
                    System.out.println(module
                        + "Processing the [submit] command");

                    tok = new StringTokenizer(s, " ");
                    int numOfTokens = tok.countTokens();
                    tok.nextToken();/* skip the command */

                    if (numOfTokens > 1) {
                        numOfMaps = Integer.parseInt(tok.nextToken());
                    }

                    deployer.prepareInstancesAndDeploy(numOfMaps);
                    continue;
                }

                if (s.startsWith("status")) {
                    System.out.println(deployer.getStatus());
                    continue;
                }

                if (s.startsWith("refresh")) {
                    String jobId = deployer.getJobId();
                    if (jobId != null) {
                        deployer.refreshStatus(jobId);
                    } else {
                        System.out.println(module
                            + "No JOB seems to have been submitted.");
                    }
                    continue;
                }

                if (s.equals("h")) {
                    System.out.println(helpInfo);
                    continue;
                }

            }/* end while(true) */

        } catch (Exception e) {
            System.out.println(module + e.getMessage());
            e.printStackTrace();
        }

    }
}
